package com.atguigu.bigdata.chapter11.function;

import com.atguigu.bigdata.bean.WaterSensor;
import lombok.AllArgsConstructor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.TableAggregateFunction;
import org.apache.flink.util.Collector;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.call;

/**
 * @Author lzc
 * @Date 2022/9/9 14:09
 */
public class Flink04_Function_TableAgg {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        
        DataStreamSource<WaterSensor> stream = env.fromElements(
            new WaterSensor("s1", 1000L, 10),
            new WaterSensor("s1", 2000L, 20),
            new WaterSensor("s1", 3000L, 30),
            new WaterSensor("s1", 4000L, 40),
            new WaterSensor("s2", 8000L, 80)
        );
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        
        Table table = tEnv.fromDataStream(stream, $("id"), $("ts"), $("vc"));
        tEnv.createTemporaryView("sensor", table);
        
        // 使用自定义函数有两种用法:
        // 1. 在table_api中使用
        // 1.1 内联用法
        
        // 1.2 先注册后使用
        tEnv.createTemporaryFunction("top2", Top2.class);
        table
            .groupBy($("id"))
            .flatAggregate(call("top2", $("vc")))
            .select($("id"), $("rank"), $("vc"))
            .execute()
            .print();
        
        // 2. 在sql中使用, 必须先注册函数
        
        
    }
    
    public static class Top2 extends TableAggregateFunction<Result, FirstSecond> {
        
        // 创建累加器
        @Override
        public FirstSecond createAccumulator() {
            return new FirstSecond();
        }
        
        // 累加过程
        public void accumulate(FirstSecond fs, Integer vc) {
            
            // 计算出来第一名和第二名
            if (vc > fs.first) {
                fs.second = fs.first; // 原来的第一名变成第二名
                fs.first = vc; // 新来的变成第一名
            } else if (vc > fs.second) {
                fs.second = vc;
            }
            
        }
        
        // 输出多行多列
        // 1. 返回值必须是 void
        // 2. 方法名必须是 emitValue
        // 参数1: 必须是累加器, 从累加器中获取最终的结果
        // 参数2: 必须是 Collector<T> 泛型必须是结果类型
        public void emitValue(FirstSecond fs, Collector<Result> out) {
            // 每来一条数据, 调用连词collect方法,意味着输出两行
            out.collect(new Result("第一名", fs.first));
            if (fs.second > Integer.MIN_VALUE) {
                // 当second有值的才向外输出. 如果等于MIN_VALUE表示没有赋值过
                out.collect(new Result("第二名", fs.second));
            }
        }
        
    }
    
    // 聚合的时候, 需要计算的是第一名和第二名的水位的值, 这个累加器只需要记住两个水位
    
    
    public static class FirstSecond {
        
        public Integer first = Integer.MIN_VALUE;
        public Integer second = Integer.MIN_VALUE;
    }
    
    // 用一个POJO来封装每行数据
    @AllArgsConstructor
    public static class Result {
        public String rank;
        public Integer vc;
    }
    
}
/*
每来一个水位, 输出每个传感器的第一和第二水位

new WaterSensor("s1", 1000L, 10),
    第一名  10
    
new WaterSensor("s1", 2000L, 20),
    第一名   20
    第二名   10
    
new WaterSensor("s1", 3000L, 30),
    第一名   30
    第二名   20
new WaterSensor("s1", 4000L, 5),
    第一名   30
    第二名   20
    
 .....
 每来一条数据, 得到两行两列: 这是什么函数的功能? TableFunction
 
 每来一条数据, 计算最大和次大: 这是什么函数的功能? AggregateFunction
 
 
 */
